package org.cassandra;

import com.alibaba.fastjson.JSONObject;
import com.datastax.oss.driver.api.core.CqlSession;
import com.datastax.oss.driver.api.core.cql.*;
import org.apache.commons.lang3.StringUtils;

import java.net.InetSocketAddress;
import java.util.Iterator;

public class DataxFetch {

    public static void main(String[] args) {

        CqlSession session = null;

        CqlSession writesession = null;
        try {
            // 连接到 Cassandra
            session = CqlSession.builder().withAuthCredentials(args[1],args[2]).withLocalDatacenter("datacenter1").withKeyspace("tyiotcloud").addContactPoint(new InetSocketAddress(args[0], 9042)).build();
            // 准备查询语句
            String query = "SELECT entity_type,entity_id,key,bool_v,dbl_v,json_v,long_v,str_v,ts FROM ts_kv_latest_cf where ts >= 1674323507000  ALLOW FILTERING";

            // 连接到 Cassandra
//            writesession = CqlSession.builder().withAuthCredentials("root","dr0VqJuy").withLocalDatacenter("datacenter1").withKeyspace("tyiotcloud").addContactPoint(new InetSocketAddress("ld-uf68jbw542qiiw534-proxy-lindorm-pub.lindorm.rds.aliyuncs.com", 9042)).build();
            writesession = CqlSession.builder().withAuthCredentials(args[4],args[5]).withLocalDatacenter("datacenter1").withKeyspace("tyiotcloud").addContactPoint(new InetSocketAddress(args[3], 9042)).build();
            // 准备插入语句
            String insert = "INSERT INTO ts_kv_latest_cf (entity_type,entity_id,key,bool_v,dbl_v,json_v,long_v,str_v,ts) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)";

            SimpleStatement simpleStatement = SimpleStatement.builder(query).setPageSize(24)
                    .build();

            ResultSet rs = session.execute(simpleStatement);

            Iterator<Row> iter = rs.iterator();
            int i = 0;
            BatchStatement batchStatement = BatchStatement.newInstance(DefaultBatchType.LOGGED);
            PreparedStatement statement = writesession.prepare(insert);
            while (!rs.isFullyFetched()) {
                rs.getAvailableWithoutFetching();
                Row row = iter.next();
                System.out.println("entity_type: " + row.getString("entity_type") + ", key: " + row.getString("key"));
                i++;
                Object data = toKvEntry(row);
                BoundStatement boundStatement = statement.bind(
                        row.get("entity_type", String.class)
                        , row.getUuid("entity_id")
                        , row.getString("key")
                        , (data instanceof Boolean) ? data : null
                        , (data instanceof Double) ? data : null
                        , (data instanceof JSONObject) ? ((JSONObject) data).toString() : null
                        , (data instanceof Long) ? data : null
                        , (data instanceof String) ? data : null
                        , row.getLong("ts"));
                writesession.executeAsync(boundStatement);
//                batchStatement.add(boundStatement);
//                if (i % 10 == 0) {
//                    // 将所有插入操作添加到批量语句中
//                    writesession.executeAsync(batchStatement);
//                    batchStatement = BatchStatement.newInstance(DefaultBatchType.LOGGED);
//                }
            }


        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (session != null) {
                session.close();
            }

            if (writesession != null) {
                writesession.close();
            }
        }
    }

    public static Object toKvEntry(Row row) {
        String strV = row.get(ModelConstants.STRING_VALUE_COLUMN, String.class);
        if (strV != null) {
            return strV;
        } else {
            Long longV = row.get(ModelConstants.LONG_VALUE_COLUMN, Long.class);
            if (longV != null) {
                return longV;
            } else {
                Double doubleV = row.get(ModelConstants.DOUBLE_VALUE_COLUMN, Double.class);
                if (doubleV != null && doubleV > 0) {
                    return doubleV;
                } else {
                    Boolean boolV = row.get(ModelConstants.BOOLEAN_VALUE_COLUMN, Boolean.class);
                    if (boolV != null) {
                        return boolV;
                    } else {
                        String jsonV = row.get(ModelConstants.JSON_VALUE_COLUMN, String.class);
                        if (StringUtils.isNotBlank(jsonV)) {
                            return JSONObject.parseObject(jsonV);
                        }
                    }
                }
            }
        }
        return null;
    }

    private static void printResults(ResultSet resultSet) {
        for (Row row : resultSet) {
            System.out.println("entity_type: " + row.getString("entity_type") + ", key: " + row.getString("key"));
        }
    }

}
